package com.ww.hadoop.mr.topn_mapjoin;

import lombok.extern.log4j.Log4j;
import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.util.Iterator;

@Log4j
public class TReducer extends Reducer<TKey, IntWritable, Text, IntWritable> {
    Text rkey = new Text();
    IntWritable rval = new IntWritable();
    @Override
    protected void reduce(TKey key, Iterable<IntWritable> values, Reducer<TKey, IntWritable, Text, IntWritable>.Context context) throws IOException, InterruptedException {
        //key           value
        //1970-6-4 33   33
        // 1970-6-4 32   32
        // 1970-6-22 31   31
        // 1970-6-4 22   22
        Iterator<IntWritable> iterator = values.iterator();
        int flag =0;
        int day = 0;
        while (iterator.hasNext()){
            log.info("key:"+key);
            iterator.next();
            log.info("key2:"+key);
            log.info(key.getYear()+"-"+key.getMonth()+"-"+key.getDay()+ " " +key.getWd()+ " " + key.getAddress());
            if (flag==0){
                rkey.set(key.getYear()+"-"+key.getMonth()+"-"+key.getDay() + " " + key.getAddress());
                rval.set(key.getWd());
                flag++;
                day = key.getDay();
                context.write(rkey,rval);
            }else if(day!=key.getDay()){
                rkey.set(key.getYear()+"-"+key.getMonth()+"-"+key.getDay()+ " " + key.getAddress());
                rval.set(key.getWd());
                context.write(rkey,rval);
                break;
            }

        }
    }
}
